package com.kingsoft.dc.khaos.module.spark.preprocess.specific.quality

import com.kingsoft.dc.khaos.module.spark.metadata.preprocess.specific.quality.OperatorFilter
import com.kingsoft.dc.khaos.util.Logging
import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.{DataFrame, Dataset, Row}

import scala.collection.mutable.ArrayBuffer

/**
  * 取值范围检核
  *
  *
  */
class RangeValueCheck extends Logging with Serializable {
  def analysisData(data: DataFrame, operatorFilter: OperatorFilter, field: String): DataFrame = {
    val condition = operatorFilter.condition.get
    //条件and、or
    val function = operatorFilter.function.get //函数
    val expressions: List[String] = operatorFilter.expressions
//    val length = expressions.length
    val df: DataFrame = data
//    val schemaFields = df.schema.fields
    val encoder = RowEncoder(df.schema)
    val dataStatistics = new DataStatistics
    val errorDataFrame: Dataset[Row] = df.flatMap(line => {
      val errorArray = ArrayBuffer[Row]()
      var record = 0
      try {
        expressions.foreach(ex => {
          val strings = ex.split(" ")
          // upd start by lianghuaxin #11755 20210628
//          if (line.getAs(field) != null && function == "" || function == null) {
          if (function == null || function == "") {
            //如不选处理函数就需要判断数据类型后进行处理了
            //            val lenValue = line.getAs(field).toString.toInt
            val lenValue = if (line.getAs(field) == null)  0 else line.getAs(field).toString.toInt
            if (strings(2) equals ("in")) {
              val list = strings(3).split(",")
              if (!list.contains(lenValue))
                record += 1
            } else {
              val i = dataStatistics.compareValue(strings, lenValue)
              record += i
            }
          }
//          if (function.equals("trim") && line.getAs(field) != null) {
          else if (function.equals("trim")) {
            //只有等于、不等于、存在
//            val fieldValue = line.getAs(field).toString.trim //trim后字段值
            val fieldValue = if (line.getAs(field) == null)  "" else line.getAs(field).toString.trim //trim后字段值
            strings(2) match {
              case "in" => {
                val list = strings(3).split(",")
                if (!list.contains(fieldValue))
                  record += 1
              }
              case "=" => {
                if (!strings(3).equals(fieldValue)) record += 1
              }
              case "!=" => {
                if (strings(3).equals(fieldValue)) record += 1
              }
            }
          }
//          if (function.equals("len") && line.getAs(field) != null) {
          else if (function.equals("len")) {
            //暂时不需要修改
//            val lenValue = line.getAs(field).toString.length //length后字段值
            val lenValue = if (line.getAs(field) == null)  0 else line.getAs(field).toString.length //length后字段值
            if (strings(2) equals ("in")) {
              val list = strings(3).split(",")
              if (!list.contains(lenValue))
                record += 1
            } else {
              val i = dataStatistics.compareValue(strings, lenValue)
              record += i
            }
          }

        })
      } catch {
        case ex: Exception => {
          ex.printStackTrace()
        }
      }
      if (condition equals ("and")) {
        if (record > 0) errorArray += line
      }
      if (condition equals ("or")) {
        if (record == expressions.length) errorArray += line
      }
      errorArray
    })(encoder)
    errorDataFrame
  }
}
